package com.cmb.lk50.sink.kafka;

import com.cmb.lk50.pojo.AbstractLog;
import com.cmb.lk50.sink.LogSink;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2020/7/29
 * Copyright    (C) ChinaMerchantsBank
 */
public class KafkaSink implements LogSink {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSink.class);

    private KafkaProducer<String, String> producer;

    private String topic;

    public KafkaSink(KafkaProducer<String, String> producer, String topic) {
        this.producer = producer;
        this.topic = topic;
    }

    @Override
    public void write(AbstractLog log) {
        writeWithKey(log, null);
    }

    public void writeWithKey(AbstractLog log, KafkaKeyGenerator<AbstractLog> keyExtractor) {
        producer.send(new ProducerRecord<>(log.kafkaTopic() == null ? topic : log.kafkaTopic(), keyExtractor == null ? null : keyExtractor.getRecordKey(log),
            gson.toJson(log, log.getClass())),
          (metadata, exception) -> {
              if (exception != null && LOGGER.isDebugEnabled()) {
                  LOGGER.debug("日志写入kafka异常, [{}]", exception.getMessage());
              }
          });
    }

}
